package com.tyrone.hephaestus.broker.register;

import com.tyrone.hephaestus.broker.entity.Repository;
import com.tyrone.hephaestus.common.annotation.Listener;
import com.tyrone.hephaestus.common.entity.Msg;
import com.tyrone.hephaestus.consumer.Consumer;
import lombok.SneakyThrows;
import org.apache.commons.lang.ObjectUtils;
import org.reflections.Reflections;

import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class Registry {

    private static HashMap<String, Set<Consumer>> channelGroup = new HashMap<>();

    {
        Class<?>[] declaredClasses = Consumer.class.getDeclaredClasses();

    }

    public static void register(String topic, Consumer consumer) {
        Set<Consumer> consumers = channelGroup.get(topic);
        if (consumers == null) {
            HashSet<Consumer> group = new HashSet<>();
            group.add(consumer);
            channelGroup.put(topic, group);
        } else {
            consumers.add(consumer);
        }
    }

    public static void boardCast(Msg msg) {
        String topic = msg.getTopic();
        Set<Consumer> consumers = channelGroup.get(topic);
        consumers.parallelStream().forEach(consumer -> consumer.onMessage(msg));
        Repository.getRepository().loss();
    }


    public static void init() {
//        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
//            Reflections reflections = new Reflections("com.tyrone");
//            Set<Class<? extends Consumer>> subTypesOf = reflections.getSubTypesOf(Consumer.class);
//            subTypesOf.forEach(subClass -> {
//                Listener listener = subClass.getAnnotation(Listener.class);
//                if (listener != null) {
//                    String topic = listener.topic();
//                    try {
//                        register(topic, subClass.newInstance());
//                    }catch (Exception e){
//                        e.getMessage();
//                    }
//                }
//            });
//        }));
        Reflections reflections = new Reflections("com.tyrone");
        Set<Class<? extends Consumer>> subTypesOf = reflections.getSubTypesOf(Consumer.class);
        subTypesOf.forEach(subClass -> {
            Listener listener = subClass.getAnnotation(Listener.class);
            if (listener != null) {
                String topic = listener.topic();
                try {
                    register(topic, subClass.newInstance());
                } catch (Exception e) {
                    e.getMessage();
                }
            }
        });
    }


}
